package com.zhss.microservice.server.node.ha;

import com.alibaba.fastjson.JSONObject;
import com.zhss.microservice.server.config.Configuration;
import com.zhss.microservice.server.constant.MessageType;
import com.zhss.microservice.server.node.Controller;
import com.zhss.microservice.server.node.ControllerCandidate;
import com.zhss.microservice.server.node.ControllerNode;
import com.zhss.microservice.server.node.ServerNodeRole;
import com.zhss.microservice.server.node.network.RemoteServerNode;
import com.zhss.microservice.server.node.network.RemoteServerNodeManager;
import com.zhss.microservice.server.node.network.ServerNetworkManager;
import com.zhss.microservice.server.slot.SlotManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 高可用管理组件
 *
 * 可伸缩架构，往集群里加入新的节点，之前崩溃了一个controller或者是candidate，你新加入了一个节点
 * 此时要看一下，他是否是controller候选节点，如果是的话，此时应该自动加入controller候选节点集群，自动同步集群元数据
 * 如果是普通节点的话，那么此时就是自动的去做数据rebalanc就可以了，自动的把槽位数据做一个重新计算，做一个重平衡
 *
 */
public class HighAvailabilityManager {

    public HighAvailabilityManager() {
        new Worker().start();
    }

    static class Singleton {
        static HighAvailabilityManager instance = new HighAvailabilityManager();
    }

    public static HighAvailabilityManager getInstance() {
        return Singleton.instance;
    }

    private LinkedBlockingQueue<Integer> disconnectedNodeQueue =
            new LinkedBlockingQueue<>();

    /**
     * 处理跟远程节点网络连接断开的异常
     * @param remoteNodeId
     */
    public void handleDisconnectedException(Integer remoteNodeId) {
        disconnectedNodeQueue.offer(remoteNodeId);
    }

    class Worker extends Thread {

        Logger LOGGER = LoggerFactory.getLogger(Worker.class);

        @Override
        public void run() {
            while(true) {
                try {
                    Integer myNodeId = Configuration.getInstance().getNodeId();
                    Integer disconnectedNodeId = disconnectedNodeQueue.take();

                    RemoteServerNodeManager remoteServerNodeManager = RemoteServerNodeManager.getInstance();
                    RemoteServerNode remoteServerNode = remoteServerNodeManager.getRemoteServerNode(disconnectedNodeId);
                    boolean isDisconnectedCandidate = remoteServerNode.isControllerCandidate();
                    String disconnectedAddress = remoteServerNode.getIp() + ":" + remoteServerNode.getClientPort();

                    releaseNetworkResource(disconnectedNodeId);

                    // 判断是否是跟controller的连接断开
                    // 如果是controller崩溃，以及自己是candidate，则必须启用高可用机制，发起controller重新选举
                    if(ServerNodeRole.isCandidate() &&
                            ControllerNode.isControllerNode(disconnectedNodeId)) {
                        ControllerCandidate controllerCandidate = ControllerCandidate.getInstance();
                        Integer serverNodeRole = controllerCandidate.electController();
                        ServerNodeRole.setRole(serverNodeRole);
                        LOGGER.info("controller重新选举后的结果：" + (serverNodeRole == ServerNodeRole.CONTROLLER ? "Controller" : "Controller候选节点"));

                        if(serverNodeRole.equals(ServerNodeRole.CONTROLLER)) {
                            recalculateAllocation(disconnectedNodeId);
                            resyncAllocation();
                        } else if(serverNodeRole.equals(ServerNodeRole.CANDIDATE)) {
                            // 如果是controller后节点，是否需要重新去接收所有的槽位和槽位副本的计算
                            // 是否需要等待重复接收一些信息这样子
                            controllerCandidate.waitForSlotsAllocation();
                            controllerCandidate.waitForSlotsReplicaAllocation();
                            controllerCandidate.waitReplicaNodeIds();
                        }

                        if(myNodeId > disconnectedNodeId) {
                            new CandidateReconnectThread(disconnectedNodeId, disconnectedAddress).start();
                        }
                    } else if(ServerNodeRole.isController()) {
                        // controller重新选举是肯定不用了
                        // 重新计算槽位、槽位副本的数据
                        recalculateAllocation(disconnectedNodeId);
                        resyncAllocation();
                        // 在重新计算的过程中，就需要把新计算好的数据同步给对应的candidate和普通节点
                        // 槽位副本的转正，槽位数据的重新分配

                        if(isDisconnectedCandidate) {
                            if(myNodeId > disconnectedNodeId) {
                                new CandidateReconnectThread(disconnectedNodeId, disconnectedAddress).start();
                            }
                        }
                    } else if(ServerNodeRole.isCandidate() &&
                            !ControllerNode.isControllerNode(disconnectedNodeId)) {
                        ControllerCandidate controllerCandidate = ControllerCandidate.getInstance();
                        controllerCandidate.waitForSlotsAllocation();
                        controllerCandidate.waitForSlotsReplicaAllocation();
                        controllerCandidate.waitReplicaNodeIds();

                        if(isDisconnectedCandidate) {
                            if(myNodeId > disconnectedNodeId) {
                                new CandidateReconnectThread(disconnectedNodeId, disconnectedAddress).start();
                            }
                        }
                    } else if(ServerNodeRole.isCommonNode() && isDisconnectedCandidate) {
                        new CandidateReconnectThread(disconnectedNodeId, disconnectedAddress).start();
                    }

                    // 通知原controller的slots槽位副本所在节点，把副本进行转正
                    // 还必须去更新自己的集群元数据，同步集群元数据，此时才可以开放元数据的使用
                } catch(Exception e) {
                    LOGGER.error("高可用机制后台线程运行出错......", e);
                }
            }

        }

        private void releaseNetworkResource(Integer disconnectedNodeId) {
            ServerNetworkManager serverNetworkManager = ServerNetworkManager.getInstance();
            serverNetworkManager.shutdownIOThread(disconnectedNodeId);
            serverNetworkManager.remoteRemoteNodeSocket(disconnectedNodeId);

            RemoteServerNodeManager remoteServerNodeManager = RemoteServerNodeManager.getInstance();
            remoteServerNodeManager.removeServerNode(disconnectedNodeId);
        }

        /**
         * 进行分配数据的重计算
         * @param disconnectedNodeId
         */
        private void recalculateAllocation(Integer disconnectedNodeId) {
            // 如果说有一个人成为了新的controller之后，那么他必须要干一些什么事情呢？
            // 他按理说就应该把之前老的controller干过的事情，都重新干一遍
            // 槽位和槽位副本的计算，是否需要重新计算一遍？是否需要重新同步给别的节点，同步完了以后是否需要重新内存进行初始化
            // 是否有新的信息需要重复讲进行发送

            // 对槽位和副本相关数据进行重计算
            ControllerCandidate controllerCandidate = ControllerCandidate.getInstance();
            Map<Integer, List<String>> slotsAllocation = controllerCandidate.getSlotsAllocation();
            Map<Integer, List<String>> slotsReplicaAllocation = controllerCandidate.getSlotsReplicaAllocation();
            Map<Integer, Integer> replicaNodeIds = controllerCandidate.getReplicaNodeIds();

            // 1、对槽位分配数据进行重计算
            List<String> disconnectedSlots = slotsAllocation.get(disconnectedNodeId);
            slotsAllocation.remove(disconnectedNodeId);

            Integer replicaNodeId = replicaNodeIds.get(disconnectedNodeId);
            List<String> replicaNodeSlots = slotsAllocation.get(replicaNodeId);
            replicaNodeSlots.addAll(disconnectedSlots);

            // 直接就是通知副本节点，把他上面的槽位副本进行转正
            changeReplicaToSlots(replicaNodeId, disconnectedSlots);

            // 2、对副本节点id进行重计算
            List<Integer> nodeIds = new ArrayList<Integer>();

            RemoteServerNodeManager remoteServerNodeManager = RemoteServerNodeManager.getInstance();
            List<RemoteServerNode> remoteMasterNodes =
                    remoteServerNodeManager.getRemoteServerNodes();
            for(RemoteServerNode remoteServerNode : remoteMasterNodes) {
                nodeIds.add(remoteServerNode.getNodeId());
            }

            nodeIds.remove(disconnectedNodeId);

            Map<Integer, Integer> newReplicaNodeIds = new HashMap<Integer, Integer>();

            for(Integer nodeId : replicaNodeIds.keySet()) {
                Integer oldReplicaNodeId = replicaNodeIds.get(nodeId);

                if(oldReplicaNodeId.equals(disconnectedNodeId)) {
                    Integer newReplicaNodeId = null;
                    boolean hasDecidedReplicaNode = false;

                    while(!hasDecidedReplicaNode) {
                        newReplicaNodeId = nodeIds.get(new Random().nextInt(nodeIds.size()));
                        if(!newReplicaNodeId.equals(nodeId)) {
                            hasDecidedReplicaNode = true;
                        }
                    }

                    newReplicaNodeIds.put(nodeId, newReplicaNodeId);
                }
            }

            for(Integer nodeId : newReplicaNodeIds.keySet()) {
                Integer newReplicaNodeId = newReplicaNodeIds.get(nodeId);
                replicaNodeIds.put(nodeId, newReplicaNodeId);
            }

            replicaNodeIds.remove(disconnectedNodeId);

            // 把有变化的副本节点id同步给对应的节点去进行重置
            for(Integer nodeId : newReplicaNodeIds.keySet()) {
                Integer newReplicaNodeId = newReplicaNodeIds.get(nodeId);
                refreshReplicaNodeId(nodeId, newReplicaNodeId);
            }

            // 3、对槽位副本数据进行重计算
            slotsReplicaAllocation = new ConcurrentHashMap<>();

            for(Integer nodeId : replicaNodeIds.keySet()) {
                Integer _replicaNodeId = replicaNodeIds.get(nodeId);
                List<String> slots = slotsAllocation.get(nodeId);
                slotsReplicaAllocation.put(_replicaNodeId, slots);
                // 执行刷新操作
                refreshReplicaSlots(_replicaNodeId, slots);
            }

            // 要在这里，去针对你的槽位副本，槽位数据所在的目标节点，把数据拉取过来，做一次正式的副本数据的同步
            // 就是并不是所有的槽位数据都要拉取和同步的，因为目标节点，也不是所有的槽位里都有数据的
            // 主要把有数据的槽位，一个一个的拉取和同步过来，就可以了
            // 对于我们现在而言，我们的槽位里的数据，主要是服务注册和心跳的数据，刚开始新的槽位副本可能是空的，我们即使不
            // 去做这个同步，也可以啊，对于那些继续在运作的节点和正常的槽位，已经注册过的会持续的进行心跳，没注册过的会继续进行注册

            LOGGER.info("重新计算后的槽位分配数据：" + slotsAllocation);
            LOGGER.info("重新计算后的副本节点id数据：" + replicaNodeIds);
            LOGGER.info("重新计算后的槽位副本数据：" + slotsReplicaAllocation);

            // 他已经转为了一个Controller了，此时他的角色应该转变为Controller，而不是ControllerCandidate
            Controller controller = Controller.getInstance();
            controller.setSlotsAllocation((ConcurrentHashMap)slotsAllocation);
            controller.setReplicaNodeIds((ConcurrentHashMap)replicaNodeIds);
            controller.setSlotsReplicaAllocation((ConcurrentHashMap)slotsReplicaAllocation);
        }

        /**
         * 重新同步分配数据给其他的controller candidate
         */
        private void resyncAllocation() {
            Controller controller = Controller.getInstance();
            controller.syncSlotsAllocation();
            controller.syncReplicaNodeIds();
            controller.syncSlotsReplicaAllocation();
        }

        /**
         * 让副本进行转正
         */
        private void changeReplicaToSlots(Integer nodeId, List<String> slots) {
            Configuration configuration = Configuration.getInstance();
            Integer myNodeId = configuration.getNodeId();

            if(nodeId.equals(myNodeId)) {
                SlotManager slotManager = SlotManager.getInstance();
                slotManager.changeReplicaToSlots(slots);
                return;
            }

            byte[] bytes = JSONObject.toJSONString(slots).getBytes();

            ByteBuffer byteBuffer = ByteBuffer.allocate(4 + bytes.length);
            byteBuffer.putInt(MessageType.CHANGE_REPLICA_TO_SLOTS);
            byteBuffer.put(bytes);

            ServerNetworkManager serverNetworkManager = ServerNetworkManager.getInstance();
            serverNetworkManager.sendMessage(nodeId, byteBuffer);
        }

        private void refreshReplicaNodeId(Integer nodeId, Integer newReplicaNodeId) {
            Configuration configuration = Configuration.getInstance();
            Integer myNodeId = configuration.getNodeId();

            if(nodeId.equals(myNodeId)) {
                SlotManager slotManager = SlotManager.getInstance();
                slotManager.refreshReplicaNodeId(newReplicaNodeId);
                return;
            }

            ByteBuffer buffer = ByteBuffer.allocate(4 + 4);
            buffer.putInt(MessageType.REFRESH_REPLICA_NODE_ID);
            buffer.putInt(newReplicaNodeId);

            ServerNetworkManager serverNetworkManager = ServerNetworkManager.getInstance();
            serverNetworkManager.sendMessage(nodeId, buffer);
        }


        private void refreshReplicaSlots(Integer nodeId, List<String> replicaSlots) {
            Configuration configuration = Configuration.getInstance();
            Integer myNodeId = configuration.getNodeId();

            if(nodeId.equals(myNodeId)) {
                SlotManager slotManager = SlotManager.getInstance();
                slotManager.refreshReplicaSlots(replicaSlots);
                return;
            }

            byte[] replicaSlotsBytes = JSONObject.toJSONString(replicaSlots).getBytes();
            ByteBuffer buffer = ByteBuffer.allocate(4 + replicaSlotsBytes.length);
            buffer.putInt(MessageType.REFRESH_REPLICA_SLOTS);
            buffer.put(replicaSlotsBytes);

            ServerNetworkManager serverNetworkManager = ServerNetworkManager.getInstance();
            serverNetworkManager.sendMessage(nodeId, buffer);
        }

    }

}
